import kafka.consumer.Consumer;
import kafka.consumer.ConsumerConfig;
import kafka.consumer.KafkaStream;
import kafka.javaapi.consumer.ConsumerConnector;
import kafka.message.MessageAndMetadata;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;

/**
 * Created by Administrator on 2016/3/17.
 */
public class KafkaConsumer extends Thread {

    private static final Properties props = new Properties();

    static {
        props.put("zookeeper.connect", KafkaConfig.ZOOKEEPER_CONNECT);
        props.put("zookeeper.session.timeout.ms", KafkaConfig.ZOOKEEPER_SESSION_TIMEOUT);
        props.put("zookeeper.sync.time.ms", KafkaConfig.ZOOKEEPER_SYNC_TIME);
        props.put("auto.commit.interval.ms", KafkaConfig.AUTO_COMMIT_INTERVAL_MS);
        props.put("group.id", KafkaConfig.GROUP_ID);
    }

    private final ConsumerConnector consumer;
    private final String topic;

    public KafkaConsumer(String topic) {
        consumer = Consumer.createJavaConsumerConnector(new ConsumerConfig(props));
        this.topic = topic;
    }

    @Override
    public void run() {
        Map<String, Integer> topicCountMap = new HashMap<String, Integer>();
        topicCountMap.put(topic, 1);
        Map<String, List<KafkaStream<byte[], byte[]>>> consumerMap = consumer.createMessageStreams(topicCountMap);
        KafkaStream<byte[], byte[]> streams = consumerMap.get(topic).get(0);
        for (MessageAndMetadata<byte[], byte[]> stream : streams) {
            System.out.println("receive：" + new String(stream.message()));
            try {
                sleep(3000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }

    }

}
